Source code for hysop.backend.hardware.cpu
# Copyright (c) HySoP 2011-2024
#
# This file is part of HySoP software.
# See "https://particle_methods.gricad-pages.univ-grenoble-alpes.fr/hysop-doc/"
# for further info.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from hysop.backend.hardware.hwinfo import TopologyObject
[docs]
class ProcessingUnit(TopologyObject):
"""
The smallest physical execution unit that hwloc recognizes.
For example, there may be multiple PUs on a core (e.g., hardware threads).
"""
def __init__(self, parent, pu):
super().__init__(parent, pu)
def _parsed_type(self):
return "PU"
def _parse_object(self, it):
raise ValueError(f"Unknown object type {_type}.")
[docs]
class PhysicalCore(TopologyObject):
"""
A single, physical processing unit which may still contain multiple logical processors,
such as hardware threads.
"""
def __init__(self, parent, core):
self._processing_units = []
super().__init__(parent, core)
[docs]
def processing_units(self):
return sorted(self._processing_units, key=lambda x: x.os_index())
[docs]
def processing_units_count(self):
return len(self._processing_units)
def _parsed_type(self):
return "Core"
def _parse_object(self, it):
_type = it.attrib["type"]
if _type == "PU":
obj = ProcessingUnit(self, it)
self._processing_units.append(obj)
else:
raise ValueError(f"Unknown object type {_type}.")
def __str__(self):
return f"core {self.os_index():>2}: {self.cpu_mask()}"
[docs]
class CpuPackage(TopologyObject):
"""
A physical package or chip, that goes into a package,
it is a grouping of one or more processors.
"""
def __init__(self, parent, package):
self._physical_cores = []
super().__init__(parent, package)
[docs]
def physical_cores(self):
return sorted(self._physical_cores, key=lambda x: x.os_index())
[docs]
def family_number(self):
return self.attribute("cpu_family_number", 0, int)
[docs]
def stepping(self):
return self.attribute("cpu_stepping", 0, int)
[docs]
def vendor(self):
return self.attribute("cpu_vendor")
[docs]
def model(self):
return self.attribute("cpu_model")
[docs]
def physical_cores_count(self):
return len(self._physical_cores)
[docs]
def processing_units_count(self):
return sum(x.processing_units_count() for x in self._physical_cores)
def __str__(self):
header = f"::Package {self.os_index()}::"
content = """
vendor: {}
model: {}
family: {}
stepping: {}
phys. cores: {}
proc. units: {}
cpuset: {}
""".format(
self.vendor(),
self.model(),
self.family_number(),
self.stepping(),
self.physical_cores_count(),
self.processing_units_count(),
self.cpu_mask(),
)
for core in self.physical_cores():
content += str(core) + "\n"
return header + self.indent(content)
def _parsed_type(self):
return "Package"
def _parse_object(self, it):
_type = it.attrib["type"]
if _type == "Core":
obj = PhysicalCore(self, it)
self._physical_cores.append(obj)
else:
raise ValueError(f"Unknown object type {_type}.")